package com.small.nacos.client.naming.core;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.small.nacos.client.naming.beat.BeatInfo;
import com.small.nacos.client.naming.beat.BeatReactor;
import com.small.nacos.client.naming.net.NamingProxy;
import com.small.nacos.client.naming.utils.UtilAndComs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;

/**
 * @Author zhoujin
 * @Date 2022/10/13 13:52
 */
public class HostReactor {

    private static final long DEFAULT_DELAY = 1000L;

    private static final long UPDATE_HOLD_INTERVAL = 5000L;
    // 服务更新任务  服务唯一标识与服务信息ServiceInfo的映射关系。服务唯一标识=groupName+@@+serviceName+@@+clusterName，其中clusterName部分可以为空
    private final Map<String, ScheduledFuture<?>> futureMap = new HashMap<String, ScheduledFuture<?>>();
    // 服务注册表  存储正在执行更新操作的service唯一标识
    private final Map<String, ServiceInfo> serviceInfoMap = new ConcurrentHashMap<>();
    // 服务更新表  存储服务唯一标识与服务更新Future的映射关系
    private final Map<String, Object> updatingMap = new ConcurrentHashMap<>();

    private NamingProxy serverProxy;

    private BeatReactor beatReactor;

    private final ScheduledExecutorService executor;

    private final Logger LOGGER = LoggerFactory.getLogger(HostReactor.class);

    public HostReactor(NamingProxy serverProxy, BeatReactor beatReactor) {
        this.executor = new ScheduledThreadPoolExecutor(UtilAndComs.DEFAULT_POLLING_THREAD_COUNT, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.small.nacos.client.naming.updater");
                return thread;
            }
        });

        this.beatReactor = beatReactor;
        this.serverProxy = serverProxy;
    }


    public ServiceInfo getServiceInfoDirectlyFromServer(String serviceName, String clusters) throws NacosException {
        String result = serverProxy.queryList(serviceName, clusters);
        if (StringUtils.isNotEmpty(result)) {
            return JacksonUtils.toObj(result, ServiceInfo.class);
        }
        return null;
    }

    public ServiceInfo getServiceInfo(String serviceName, String clusters) {
        ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);

        if (serviceObj == null) {

            serviceObj = new ServiceInfo(serviceName, clusters);

            serviceInfoMap.put(serviceObj.getKey(), serviceObj);

            updatingMap.put(serviceName, serviceObj);
            updateServiceNow(serviceName, clusters);
            updatingMap.remove(serviceName);

        } else if (updatingMap.containsKey(serviceName)) {

            if (UPDATE_HOLD_INTERVAL > 0){
                synchronized (serviceObj) {
                    try {
                        serviceObj.wait(UPDATE_HOLD_INTERVAL);
                    } catch (InterruptedException e) {
                        LOGGER
                                .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
                    }
                }
            }
        }

        //添加定时任务
        scheduleUpdateIfAbsent(serviceName, clusters);

        return serviceInfoMap.get(serviceObj.getKey());
    }

    private void updateServiceNow(String serviceName, String clusters) {
        try {
            updateService(serviceName, clusters);
        } catch (NacosException e) {
            LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
        }
    }

    private void updateService(String serviceName, String clusters) throws NacosException {
        ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
        try {
            String result = serverProxy.queryList(serviceName, clusters);

            if (StringUtils.isNotEmpty(result)) {
                processServiceJson(result);
            }
        } finally {
            if (oldService != null) {
                //更新完唤醒其他等待更新的线程
                synchronized (oldService) {
                    oldService.notifyAll();
                }
            }
        }

    }

    /**
     * Refresh only.
     *
     * @param serviceName service name
     * @param clusters    cluster
     */
    public void refreshOnly(String serviceName, String clusters) {
        try {
            serverProxy.queryList(serviceName, clusters);
        } catch (Exception e) {
            LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
        }
    }

    private ServiceInfo processServiceJson(String json) {

        if (StringUtils.isBlank(json)) {
            return null;

        }
        ServiceInfo serviceInfo = JacksonUtils.toObj(json, ServiceInfo.class);

        String key = serviceInfo.getKey();
        if (StringUtils.isBlank(key)) {
            return null;
        }
        ServiceInfo oldServiceInfo = serviceInfoMap.get(key);
        serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
        serviceInfo.setJsonFromServer(json);

        Boolean changed = false;
        if (oldServiceInfo != null) {

            List<Instance> oldHosts = oldServiceInfo.getHosts();
            HashMap<String, Instance> oldHostsHashMap = new HashMap();
            for (Instance host : oldHosts) {
                oldHostsHashMap.put(host.toInetAddr(), host);
            }

            List<Instance> hosts = serviceInfo.getHosts();
            HashMap<String, Instance> hostsHashMap = new HashMap();
            for (Instance host : hosts) {
                hostsHashMap.put(host.toInetAddr(), host);
            }

            Set<Instance> modHosts = new HashSet<Instance>();
            Set<Instance> newHosts = new HashSet<Instance>();
            Set<Instance> remvHosts = new HashSet<Instance>();

            for (Map.Entry<String, Instance> hostEntry : hostsHashMap.entrySet()) {
                if (oldHostsHashMap.containsKey(hostEntry.getKey()) && !oldHostsHashMap.get(hostEntry.getKey()).toString().equals(hostEntry.getValue().toString())) {
                    modHosts.add(hostEntry.getValue());
                }

                if (!oldHostsHashMap.containsKey(hostEntry.getKey())) {
                    newHosts.add(hostEntry.getValue());
                }
            }


            for (Map.Entry<String, Instance> oldInstanceEntry : oldHostsHashMap.entrySet()) {
                if (!hostsHashMap.containsKey(oldInstanceEntry.getKey())) {

                    remvHosts.add(oldInstanceEntry.getValue());
                }
            }

            if (newHosts.size() > 0) {
                changed = true;
                LOGGER.info("new ips(" + newHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
                        + JacksonUtils.toJson(newHosts));
            }

            if (remvHosts.size() > 0) {
                changed = true;
                LOGGER.info("removed ips(" + remvHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
                        + JacksonUtils.toJson(remvHosts));
            }
            if (modHosts.size() > 0) {
                changed = true;
                updateBeatInfo(modHosts);
                LOGGER.info("modified ips(" + modHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
                        + JacksonUtils.toJson(modHosts));
            }


        } else {
            changed = true;
            LOGGER.info("init new ips(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> "
                    + JacksonUtils.toJson(serviceInfo.getHosts()));
        }

        if (changed) {
            /*NotifyCenter.publishEvent(new InstancesChangeEvent(this.notifierEventScope, serviceInfo.getName(), serviceInfo.getGroupName(),
                    serviceInfo.getClusters(), serviceInfo.getHosts()));
            DiskCache.write(serviceInfo, cacheDir);*/
        }

        return serviceInfo;
    }

    private void updateBeatInfo(Set<Instance> modHosts) {
        for (Instance instance : modHosts) {
            String key = beatReactor.buildKey(instance.getServiceName(), instance.getIp(), instance.getPort());
            if (beatReactor.dom2Beat.containsKey(key) && instance.isEphemeral()) {
                BeatInfo beatInfo = beatReactor.buildBeatInfo(instance);
                beatReactor.addBeatInfo(instance.getServiceName(), beatInfo);
            }
        }
    }

    private void scheduleUpdateIfAbsent(String serviceName, String clusters) {

        String key = ServiceInfo.getKey(serviceName, clusters);
        if (!futureMap.containsKey(key)) {
            synchronized (futureMap) {
                if (!futureMap.containsKey(key)) {
                    ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
                    futureMap.put(key, future);
                }
            }
        }

    }

    private ScheduledFuture<?> addTask(UpdateTask task) {
        ScheduledFuture<?> future = executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
        return future;
    }

    private ServiceInfo getServiceInfo0(String serviceName, String clusters) {

        String key = ServiceInfo.getKey(serviceName, clusters);

        return serviceInfoMap.get(key);
    }

    class UpdateTask implements Runnable {

        long lastRefTime = Long.MAX_VALUE;
        private final String clusters;
        private final String serviceName;
        private int failCount = 0;

        public UpdateTask(String serviceName, String clusters) {
            this.serviceName = serviceName;
            this.clusters = clusters;
        }

        @Override
        public void run() {

            boolean stop = false;
            long delayTime = DEFAULT_DELAY;

            try {
                String key = ServiceInfo.getKey(serviceName, clusters);
                ServiceInfo serviceObj = serviceInfoMap.get(key);
                if (serviceObj == null) {
                    updateService(serviceName, clusters);
                    return;
                }
                if (serviceObj.getLastRefTime() <= this.lastRefTime) {
                    updateService(serviceName, clusters);
                    serviceObj = serviceInfoMap.get(key);
                } else {
                    refreshOnly(serviceName, clusters);
                }

                if (!futureMap.containsKey(key)) {
                    stop = true;
                    serviceInfoMap.remove(key);
                }

                lastRefTime = serviceObj.getLastRefTime();

                if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
                    incFailCount();
                    return;
                }

                delayTime = serviceObj.getCacheMillis();

                resetFailCount();

            } catch (Exception e) {
                incFailCount();
            } finally {
                if (!stop) {
                    executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);
                }
            }

        }

        private void incFailCount() {
            int limit = 6;
            if (failCount == limit) {
                return;
            }
            failCount++;
        }

        private void resetFailCount() {
            failCount = 0;
        }
    }


}
